package com.xahj.bd2104.spark.core

import org.apache.spark.util.{AccumulatorV2, LongAccumulator}
import org.apache.spark.{SparkConf, SparkContext}

/**
 * Created with IntelliJ IDEA.
 * Author: Amos
 * E-mail: amos@amoscloud.com
 * Date: 2021/9/15
 * Time: 11:40
 * Description: 
 */
object SparkCoreDemo10_累加器 {
  def main(args: Array[String]): Unit = {
    val sc = new SparkContext(new SparkConf()
      .setAppName("demo10")
      .setMaster("local[2]"))


    val rdd = sc.makeRDD(1 to 10)
    var count = 0

    //  Spark提供了专门用于分布式聚合的累加器
    //  1. 创建累加器对象
    val acc = new LongAccumulator
    val acc1 = new MyAcc
    //  2. 将累加器注册到sc
    sc.register(acc)
    sc.register(acc1)


    println(Thread.currentThread().getId + "driver中的count值" + count)

    // -------RDD的操作在任务执行时，由每个Worker的Task线程进行------------
    rdd.map(x => {
      count += 1
      // 3. 使用累加器对象进行自增操作
      acc.add(1)
      acc1.add(1)
      println(Thread.currentThread().getId + "map中的count值" + count)
      x + 1
    }).collect()
    // --------RDD的操作-----------


    println(Thread.currentThread().getId + "driver中的count值" + count)


    // 4. 取值
    println(acc1.value)


  }
}

//自定义累加器实现Long类型计数
//                            [输入数据类型,  累加器的返回值]
class MyAcc extends AccumulatorV2[Long, Long] {

  var count: Long = 0L

  /**
   * 判断当前累加器是否为初始状态
   *
   * @return
   */
  override def isZero: Boolean = this.count == 0

  /**
   * 复制当前累加器
   *
   * @return
   */
  override def copy(): AccumulatorV2[Long, Long] = {
    val acc = new MyAcc
    acc.count = this.count
    acc
  }

  /**
   * 重置累加器
   */
  override def reset(): Unit = {
    this.count = 0L
  }

  /**
   * 为累加器加入新值
   *
   * @param v
   */
  override def add(v: Long): Unit = this.count += v

  /**
   * 合并累加器
   *
   * @param other
   */
  override def merge(other: AccumulatorV2[Long, Long]): Unit = {
    this.count += other.value
  }

  /**
   * 获取累加器的值
   *
   * @return
   */
  override def value: Long = this.count
}
